package edu.ucla.cs.typecast.ping;

import java.io.*;
import java.net.*;
import java.util.*;

import edu.ucla.cs.typecast.net.NetUtil;
import edu.ucla.cs.typecast.net.TypeCastConstants;
import edu.ucla.cs.typecast.rmi.TypeCastClient;
import edu.ucla.cs.typecast.rmi.TypeCastServer;
import edu.ucla.cs.typecast.util.ExpiringTreeMap;
import edu.ucla.cs.typecast.util.NullComparator;
import edu.ucla.cs.typecast.util.slf.ScriptLogReader;
import edu.ucla.cs.typecast.util.slf.ScriptLogWriter;

/**
 * A ping sender.
 */

public class PingSenderImpl implements PingSender {
	private static final long serialVersionUID = 1L;

	protected static int rand() {
		return (int) (Math.random() * Integer.MAX_VALUE);
	}

	protected static int randPort() {
		return (int) (Math.random() * 60000) + 1024;
	}

	protected static void sleep(long time) {
		try {
			Thread.sleep(time);
		} catch (InterruptedException e) {
			// Ignore
		}
	}

	// mcast TC client to send pings to PingReceiver objects.

	protected TypeCastClient mcastTcClient;

	// ucast TC server to receive pongs from PingReceiver objects.

	protected TypeCastServer ucastTcServer;

	// Addresses.

	protected InetAddress privateHost;

	protected int privatePort;

	protected InetAddress publicHost;

	protected int publicPort;

	// Time that the sender waits for a reply before declaring a timeout.

	protected int timeout;

	// A reference to itself to be used by outstandingPings.

	protected final PingSenderImpl pingSender = this;

	// A mapping of <sequence #> => <# of pongs received for this seq #>.

	protected ExpiringTreeMap<Integer, Integer> outstandingPings;

	/**
	 * Creates a ping sender.
	 * 
	 * @param privateHost
	 *            Unicast hostname of this sender.
	 * @param privatePort
	 *            Unicast port of this sender.
	 * @param publicHost
	 *            Multicast hostname of the "environment".
	 * @param publicPort
	 *            Multicast port of the "environment".
	 */

	public PingSenderImpl(InetAddress privateHost, int privatePort,
			InetAddress publicHost, int publicPort, int timeout)
			throws IOException {

		// The map calls the callback functions when a sequence # is about to
		// expire (as no new pong for that seq # is received).

		outstandingPings = new ExpiringTreeMap<Integer, Integer>(
				new NullComparator(), timeout) {
			private static final long serialVersionUID = 1L;

			public void onExpireEntry(Integer seqNum, Integer receiveCount) {
				if (seqNum != null && receiveCount != null && receiveCount > 0) {
					pingSender.onExpectedTimeout(seqNum);
				} else {
					pingSender.onUnexpectedTimeout(seqNum);
				}
			}
		};

		// Record the hosts and ports

		this.privateHost = privateHost;
		this.privatePort = privatePort;
		this.publicHost = publicHost;
		this.publicPort = publicPort;

		// Record the timeout

		this.timeout = timeout;

		// Set up a mcast TC client to send pings to PingReceiver objects

		mcastTcClient = new TypeCastClient(publicHost, publicPort);

		// Set up a ucast TC server to receive pongs from PingReceiver objects

		ucastTcServer = new TypeCastServer(privateHost, privatePort);
		ucastTcServer.export(this);
	}

	/**
	 * Creates a ping sender.
	 * 
	 * @param privateHost
	 *            Unicast hostname of this receiver.
	 * @param privatePort
	 *            Unicast port of this receiver.
	 * @param publicHost
	 *            Multicast hostname of the "environment".
	 * @param publicPort
	 *            Multicast port of the "environment".
	 */

	public PingSenderImpl(String privateHost, int privatePort,
			String publicHost, int publicPort, int timeout) throws IOException {
		this(InetAddress.getByName(privateHost), privatePort, InetAddress
				.getByName(publicHost), publicPort, timeout);
	}

	/**
	 * Un-export itself so that it won't act on future datagrams.
	 */

	public void dispose() {
		ucastTcServer.unexport(this);
	}

	/**
	 * Called when timeout happens after receiving at least one pong.
	 * 
	 * @param seqNum
	 */

	public void onExpectedTimeout(int seqNum) {
		// To be overrided
	}

	/**
	 * Called when an invalid pong (i.e. a pong for a ping that we have not
	 * sent) is received.
	 * 
	 * @param senderHost
	 * @param senderPort
	 * @param seqNum
	 */

	public void onInvalidPong(InetAddress senderHost, int senderPort, int seqNum) {
		// To be overrided
	}

	/**
	 * Called when timeout happens before receiving any pong.
	 * 
	 * @param seqNum
	 */

	public void onUnexpectedTimeout(int seqNum) {
		// To be overrided
	}

	/**
	 * Called when an valid pong (i.e. a pong for a ping that we have sent) is
	 * received.
	 * 
	 * @param senderHost
	 * @param senderPort
	 * @param seqNum
	 */

	public void onValidPong(InetAddress senderHost, int senderPort, int seqNum) {
		// To be overrided
	}

	/**
	 * Send a ping to ping receivers.
	 * 
	 * @param seqNum
	 * @throws IOException
	 */

	public void ping(int seqNum) throws IOException {
		// Reset the counter for this seq #

		outstandingPings.put(seqNum, 0);

		// Send a ping to the receiver

		PingReceiver pr = (PingReceiver) mcastTcClient
				.getTypeCastInvocationHandler(PingReceiver.class);
		pr.receivePing(privateHost, privatePort, seqNum);
	}

	/**
	 * Receive a pong from ping receivers.
	 * 
	 * @param senderHost
	 * @param senderPort
	 * @param seqNum
	 * @throws IOException
	 */

	public void receivePong(InetAddress receiverHost, int receiverPort,
			int seqNum) throws IOException {
		// Got a pong
		// Check if we have record for the seq # and call the appropriate
		// callback

		if (outstandingPings.containsKey(seqNum)) {
			outstandingPings.put(seqNum, outstandingPings.get(seqNum) + 1);
			onValidPong(receiverHost, receiverPort, seqNum);
		} else {
			onInvalidPong(receiverHost, receiverPort, seqNum);
		}
	}

	/**
	 * Expire timed-out entries and call the appropriate callback functions.
	 */

	public void update() {
		outstandingPings.size();
	}
}
